package hamster.csustef.acquisition.flink.sink;

import hamster.csustef.acquisition.constant.AcquisitionConstant;
import hamster.csustef.acquisition.flink.sink.serialization.StringSerializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;

import java.util.Properties;

public class KafkaProducerSinkFactory {


    public static FlinkKafkaProducer<String> getStringKafkaProducerSink(
            String servers, String topic) {
        return getKafkaProducerSink(servers, topic, new StringSerializationSchema(topic));
    }

    public static <T> FlinkKafkaProducer<T> getKafkaProducerSink(
            String servers, String topic, KafkaSerializationSchema<T> kss) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", servers);

        return new FlinkKafkaProducer<T>(
                topic,
                kss,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        String filePath = "file:///root/data/2022_date_dime.txt";

        DataStreamSource<String> fileDS = env.readTextFile(filePath);

        fileDS.print();
        fileDS.sinkTo(HdfsSinkFactory.getHdfsSink(AcquisitionConstant.CS_WEATHER_TEMP_FILE_PATH));


        env.execute();
    }

}
